package com.j.lemon.learn.flink.dataset;

import com.j.lemon.learn.flink.WordCount;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;

import java.util.List;

/**
 * 将多个元素，组合成一个元素
 * @author lijunjun
 */
public class ReduceOperator {
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSet<WordCount> dataSet = env.fromElements(
                new WordCount("zhangsan", 1),
                new WordCount("lisi", 2),
                new WordCount("lisi", 3)).reduce(new ReduceFunction<WordCount>() {
            @Override
            public WordCount reduce(WordCount wordCount, WordCount t1) throws Exception {
                wordCount.setWord(wordCount.getWord()+t1.getWord());
                wordCount.setCount(wordCount.getCount()+t1.getCount());
                return wordCount;
            }
        });
        List<WordCount> collect =
                dataSet.collect();
        System.out.println(collect);
        dataSet.print();

    }
}
